﻿/**************************************************************
 * Copyright Eastmoney Corporation. All rights reserved.
 * 
 * Author: guitao(guitao@eastmoney.com) 
 * Create Date: 2020/6/15 13:58:28
 * Description:
 *          
 * Revision History:
 *      Date         Author               Description
 *              
***************************************************************/

using Confluent.Kafka;
using gt.EventBus.Abstractions;
using gt.EventBus.Events;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace gt.EventBus.Kafka
{
    public class EventBusKafka : IEventBus, IDisposable
    {
        private readonly IKafkaPersistentConnection _kafkaConnection;
        private readonly IEventBusSubscriptionsManager _subscriptionManager;

        public EventBusKafka(IKafkaPersistentConnection kafkaConnection,
             IEventBusSubscriptionsManager subscriptionManager)
        {
            _kafkaConnection = kafkaConnection;
            _subscriptionManager = subscriptionManager;

            StartConsumerAsync().ConfigureAwait(false);

            Thread.Sleep(1000);
        }
        public EventBusKafka(IKafkaPersistentConnection kafkaConnection)
            : this(kafkaConnection, new InMemoryEventBusSubscriptionsManager())
        {
        }


        public void Publish(IntegrationEvent @event)
        {
            var producer = _kafkaConnection.Producer ?? throw new ArgumentNullException("kafka producer");
            var eventName = @event.GetType().Name;

            //以eventName为message key
            var message = new Message<string, string> { Key = eventName, Value = JsonConvert.SerializeObject(@event) };

            producer.ProduceAsync(_kafkaConnection.TopicName, message);
        }

        public void Subscribe<T, TH>()
            where T : IntegrationEvent
            where TH : IIntegrationEventHandler<T>
        {
            _subscriptionManager.AddSubscription<T, TH>();
        }

        public void Unsubscribe<T, TH>()
            where T : IntegrationEvent
            where TH : IIntegrationEventHandler<T>
        {
            _subscriptionManager.RemoveSubscription<T, TH>();
        }

        public void Dispose()
        {
            if (_kafkaConnection != null)
            {
                _kafkaConnection.Dispose();
            }
        }

        #region Private

        private async Task StartConsumerAsync()
        {
            await Task.Run(() =>
            {
                while (true)
                {
                    try
                    {
                        var consumer = _kafkaConnection.Consumer;

                        var consumeResult = consumer.Consume();
                        var eventName = consumeResult.Message.Key;

                        var handlerTypes = _subscriptionManager.GetHandlersForEvent(eventName);
                        foreach (var handlerType in handlerTypes)
                        {
                            try
                            {
                                var method = handlerType.GetMethod("Handle");
                                var eventType = _subscriptionManager.GetEventByName(eventName);
                                var integrationEvent = JsonConvert.DeserializeObject(consumeResult.Message.Value, eventType);

                                var instance = Activator.CreateInstance(handlerType);
                                method.Invoke(instance, new object[] { integrationEvent });
                            }
                            catch (Exception ex)
                            {
                                //logging
                            }
                        }

                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine($"consumer error:{ex.Message + ex.StackTrace}");
                    }
                }

            });
        }

        #endregion

    }
}
